/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.ha;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;

/**
 * 主从同步的核心实现类
 *
 * RocketMQ 主从同步的整体工作机制大致是：
 *
 * 1.从服务器主动建立 TCP 连接主服务器，然后每隔 5s 向主服务器发送 commitLog 文件最大偏移量拉取还未同步的消息；
 * 2.主服务器开启监听端口，监听从服务器发送过来的信息，主服务器收到从服务器发过来的偏移量进行解析，并返回查找出未同步的消息给从服务器；
 * 3.客户端收到主服务器的消息后，将这批消息写入 commitLog 文件中，然后更新 commitLog 拉取偏移量，接着继续向主服务拉取未同步的消息。
 *
 * 从 HA 实现逻辑可看出，可大致分为两个过程，分别是从服务器上报偏移量，以及主服务器发送未同步消息到从服务器。
 */

public class HAService {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    /**
     * 表示有几个从broker连接到主broker的连接个数
     */
    private final AtomicInteger connectionCount = new AtomicInteger(0);

    //对应着多个HAConnection，一个HAConnection表示一个从节点连接
    private final List<HAConnection> connectionList = new LinkedList<>();

    //是个线程，主节点监听从节点的服务
    private final AcceptSocketService acceptSocketService;

    private final DefaultMessageStore defaultMessageStore;

    private final WaitNotifyObject waitNotifyObject = new WaitNotifyObject();
    //记录当前与slave同步的最大的物理偏移量
    private final AtomicLong push2SlaveMaxOffset = new AtomicLong(0);

    //是个线程
    private final GroupTransferService groupTransferService;

    //是个线程
    private final HAClient haClient;

    public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
        this.defaultMessageStore = defaultMessageStore;
        this.acceptSocketService =
            new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
        this.groupTransferService = new GroupTransferService();
        this.haClient = new HAClient();
    }

    public void updateMasterAddress(final String newAddr) {
        if (this.haClient != null) {
            this.haClient.updateMasterAddress(newAddr);
        }
    }

    public void putRequest(final CommitLog.GroupCommitRequest request) {
        //groupTransferService是个线程
        this.groupTransferService.putRequest(request);
    }

    public boolean isSlaveOK(final long masterPutWhere) {
        //判断是否有从节点已经连接到主节点上。连接个数，至少有一个就返回true
        boolean result = this.connectionCount.get() > 0;
        //this.defaultMessageStore.getMessageStoreConfig().getHaSlaveFallbehindMax() =256M
        //下面代码的意思是：
        //1.如果有或至少有一个从节点连接上主节点
        //2.如果当前主节点的消息放进来的位置减去上次主节点同步给从节点的最大位置在容许的落后于主节点的差值内，
        //则返回true
        //否则返回false
        result =
            result
                && ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore
                .getMessageStoreConfig().getHaSlaveFallbehindMax());
        return result;
    }

    /**
     * 在Master收到从服务器的拉取请求，
     * 拉取请求是slave下一次待拉取的消息偏移量，
     * 也可以认为是Slave的拉取偏移量确认信息，
     * 如果该信息大于push2SlaveMaxOffset，则更新push2SlaveMaxOffset，
     * 然后唤醒GroupTransferService线程，
      */
    public void notifyTransferSome(final long offset) {
        //offset > value说明主节点数据多于从节点数据，需要同步处理
        for (long value = this.push2SlaveMaxOffset.get(); offset > value; ) {
            //把value的值变成offet
            boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, offset);
            //如果变更成功的话，使用groupTransferService进行数据同步
            if (ok) {
                //唤醒groupTransferService这个线程，继续执行run()操作
                this.groupTransferService.notifyTransferSome();
                break;
            } else {
                //赋值继续循环
                value = this.push2SlaveMaxOffset.get();
            }
        }
    }

    public AtomicInteger getConnectionCount() {
        return connectionCount;
    }

    // public void notifyTransferSome() {
    // this.groupTransferService.notifyTransferSome();
    // }

    public void start() throws Exception {
        //acceptSocketService主服务器监听从服务器连接实现类,开始监听从服务器的连接，是个线程
        this.acceptSocketService.beginAccept();
        this.acceptSocketService.start();
        //groupTransferService主从同步通知类，实现同步复制和异步复制的功能，是个线程
        this.groupTransferService.start();
        //haClient从服务器连接主服务实现类，是个线程
        this.haClient.start();
    }

    public void addConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.add(conn);
        }
    }

    public void removeConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.remove(conn);
        }
    }

    public void shutdown() {
        this.haClient.shutdown();
        this.acceptSocketService.shutdown(true);
        this.destroyConnections();
        this.groupTransferService.shutdown();
    }

    public void destroyConnections() {
        synchronized (this.connectionList) {
            for (HAConnection c : this.connectionList) {
                c.shutdown();
            }

            this.connectionList.clear();
        }
    }

    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public WaitNotifyObject getWaitNotifyObject() {
        return waitNotifyObject;
    }

    public AtomicLong getPush2SlaveMaxOffset() {
        return push2SlaveMaxOffset;
    }

    /**
     * Listens to slave connections to create {@link HAConnection}.
     *
     * 主服务器监听从服务器连接实现类
     */
    class AcceptSocketService extends ServiceThread {
        private final SocketAddress socketAddressListen;
        private ServerSocketChannel serverSocketChannel;
        private Selector selector;

        public AcceptSocketService(final int port) {
            this.socketAddressListen = new InetSocketAddress(port);
        }

        /**
         * Starts listening to slave connections.
         *
         * @throws Exception If fails.
         */
        public void beginAccept() throws Exception {
            //主服务器服务端开启一个serverSocketChannel
            this.serverSocketChannel = ServerSocketChannel.open();
            //创建一个选择器
            this.selector = RemotingUtil.openSelector();
            //在Linux平台，只有TCP状态位于 TIME_WAIT ，才可以重用端口。这才是正确的行为。
            //setReuseAddress意思是多个socket可以复用一个端口
            this.serverSocketChannel.socket().setReuseAddress(true);
            //绑定一个地址
            this.serverSocketChannel.socket().bind(this.socketAddressListen);
            //设置为非阻塞
            this.serverSocketChannel.configureBlocking(false);
            //把这个serverSocketChannel注册到selector上
            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void shutdown(final boolean interrupt) {
            super.shutdown(interrupt);
            try {
                this.serverSocketChannel.close();
                this.selector.close();
            } catch (IOException e) {
                log.error("AcceptSocketService shutdown exception", e);
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    //选择器进行选择
                    this.selector.select(1000);
                    Set<SelectionKey> selected = this.selector.selectedKeys();

                    if (selected != null) {
                        for (SelectionKey k : selected) {
                            if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                              //自己加的。为啥不是这个代码  SocketChannel sc  = this.serverSocketChannel.accept();

                                //拿到从节点的一个SocketChannel对象
                                SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();

                                if (sc != null) {
                                    HAService.log.info("HAService receive new connection, "
                                        + sc.socket().getRemoteSocketAddress());

                                    //对从服务器连接的请求进行封装成HAConnection对象
                                    try {
                                        HAConnection conn = new HAConnection(HAService.this, sc);
                                        // this.readSocketService.start();
                                        // this.writeSocketService.start();启动了这2个线程
                                        conn.start();
                                        //把这个conn对象加到一个集合中。一个连接一个这样的对象
                                        HAService.this.addConnection(conn);
                                    } catch (Exception e) {
                                        log.error("new HAConnection exception", e);
                                        sc.close();
                                    }
                                }
                            } else {
                                log.warn("Unexpected ops in select " + k.readyOps());
                            }
                        }
                       //最后一步一定要把selected清除掉
                        selected.clear();
                    }
                } catch (Exception e) {
                    log.error(this.getServiceName() + " service has exception.", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public String getServiceName() {
            return AcceptSocketService.class.getSimpleName();
        }
    }

    /**
     * GroupTransferService Service
     *
     * GroupTransferService 是一个通知线程，用于通知等待HA同步结果而阻塞的消息发送者线程
     */
    class GroupTransferService extends ServiceThread {

        private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
        private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>();
        private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>();

        public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {
            synchronized (this.requestsWrite) {
                this.requestsWrite.add(request);
            }
            if (hasNotified.compareAndSet(false, true)) {
                //这个地方是唤醒GroupTransferService这个线程的run()方法继续执行
                waitPoint.countDown(); // notify
            }
        }

        public void notifyTransferSome() {
            this.notifyTransferObject.wakeup();
        }

        /**
         * 把requestsWrite与requestsRead做交换
         */
        private void swapRequests() {
            List<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
            this.requestsWrite = this.requestsRead;
            this.requestsRead = tmp;
        }

        /**
         * 判断主从同步是否完成的依据是：
         * 所有Slave中已成功复制的最大偏移量是否大于等于消息生产者发送消息后消息服务端返回下一条消息的起始偏移量，
         * 如果是则表示主从同步复制已经完成，唤醒消息发送线程，否则等待1s,再次判断，
         * 每一个任务在一批任务中循环判断5次。消息消费者返回有两种情况：
         * 如果等待超过5s或 GroupTransferService通知主从复制完成则返回。
         * 可以通过syncFlushTimeout来设置等待时间。
         */
        private void doWaitTransfer() {
            synchronized (this.requestsRead) {
                if (!this.requestsRead.isEmpty()) {
                    //对于提交上来的每个等待复制请求，判断是否达到
                    //预期的复制进度
                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {
                        //如果当前主节点已经同步到从节点的最大位置已经大于消息的要同步的位置
                        //则说明已经同步完成，
                        boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                        //获取同步时间,当前时间+超时时间
                        long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()
                            + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
                        //否则的话，再尝试五次判断，每次间隔一秒
                        while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
                            //那就再等1S钟
                            this.notifyTransferObject.waitForRunning(1000);
                            //循环进行判断
                            transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                        }

                       //如果五次依然失败，则记录日志
                        if (!transferOK) {
                            log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
                        }
                        //令等待同步复制的handleHA方法返回，并通过transferOK
                        //这个地方唤醒的commitlog类的856行之前执行的主线程
                        req.wakeupCustomer(transferOK);
                    }

                    //清楚这次同步的请求数据
                    this.requestsRead.clear();
                }
            }
        }

        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    //1.把当前这个线程表示为未唤醒，hasNotified=false
                    //2.阻塞当前这个线程
                    //3.将requestsWrite与requestsRead做交换
                    this.waitForRunning(10);
                    this.doWaitTransfer();
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        @Override
        protected void onWaitEnd() {
            this.swapRequests();
        }

        @Override
        public String getServiceName() {
            return GroupTransferService.class.getSimpleName();
        }
    }

    //从服务器连接主服务实现类
    class HAClient extends ServiceThread {
        //4M
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024 * 4;
        //主服务器地址
        private final AtomicReference<String> masterAddress = new AtomicReference<>();
        //上报偏移量给主节点用到的缓存对象
        private final ByteBuffer reportOffset = ByteBuffer.allocate(8);
        //从服务器的socketChannel
        private SocketChannel socketChannel;
        //主服务器的选择器
        private Selector selector;
        //上次写入时间
        private long lastWriteTimestamp = System.currentTimeMillis();

        //表示本次上传的物理偏移量
        private long currentReportedOffset = 0;
        //表示本次从buffer中读取的位置
        private int dispatchPosition = 0;
        //主节点发送消息过来用来接收内容的缓存
        private ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        //备读缓冲区
        private ByteBuffer byteBufferBackup = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

        public HAClient() throws IOException {
            //开启selector对象
            this.selector = RemotingUtil.openSelector();
        }

        //备用Broker向Name Server注册时会返回主用Broker的地址
        public void updateMasterAddress(final String newAddr) {
            //得到主服务器的地址
            String currentAddr = this.masterAddress.get();
            //如果当前的主服务器的地址为空或者不等于传进来的主服务器地址
            if (currentAddr == null || !currentAddr.equals(newAddr)) {
                //则把最的地址赋值给这个haclient的masteradress的变量
                this.masterAddress.set(newAddr);
                log.info("update master address, OLD: " + currentAddr + " NEW: " + newAddr);
            }
        }

        /**
         * 判断是否上报时间已经到了
         * @return
         */
        private boolean isTimeToReportOffset() {
            //当前时间减去上次上报时间
            long interval =
                HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp;
            //如果间隔时间大于5秒钟，说明可以上报了
            boolean needHeart = interval > HAService.this.defaultMessageStore.getMessageStoreConfig()
                .getHaSendHeartbeatInterval();

            return needHeart;
        }

        private boolean reportSlaveMaxOffset(final long maxOffset) {
            this.reportOffset.position(0);
            this.reportOffset.limit(8);
            this.reportOffset.putLong(maxOffset);
            this.reportOffset.position(0);
            this.reportOffset.limit(8);

            for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
                try {
                    //把reportOffset写到服务端
                    this.socketChannel.write(this.reportOffset);
                } catch (IOException e) {
                    log.error(this.getServiceName()
                        + "reportSlaveMaxOffset this.socketChannel.write exception", e);
                    return false;
                }
            }

            //更新最新写入时间为当前时间
            lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
            return !this.reportOffset.hasRemaining();
        }

        private void reallocateByteBuffer() {
            //READ_MAX_BUFFER_SIZE = 4M
            //检查byteBufferRead的缓存中针对当前的dispatchPosition还剩多少空间大小
            int remain = READ_MAX_BUFFER_SIZE - this.dispatchPosition;
            //若reamin>0表示还有空间存放数据
            if (remain > 0) {
                //把position的位置调整到当前的dispatchPosition位置
                this.byteBufferRead.position(this.dispatchPosition);
                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);
               //将byteBufferRead中剩下的数据复制到byteBufferBackup这个缓存中；
                this.byteBufferBackup.put(this.byteBufferRead);
            }

            //将byteBufferRead和byteBufferBackup的数据进行交换；
            //交换的目的是把已经处理过的数据备用。把没有处理过的数据交给haclient线程去处理
            this.swapByteBuffer();

            /**
             * 重新初始化byteBufferRead变量的position等于reamin，
             * 即表示byteBufferRead中写入到了位置position；
             */
            this.byteBufferRead.position(remain);
            this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);
            //把分发的位置置为0
            this.dispatchPosition = 0;
        }

        private void swapByteBuffer() {
            ByteBuffer tmp = this.byteBufferRead;
            this.byteBufferRead = this.byteBufferBackup;
            this.byteBufferBackup = tmp;
        }

        private boolean processReadEvent() {
            //若为空将重复读取3次后仍然没有数据则跳出该循环
            int readSizeZeroTimes = 0;
             //该方法的逻辑为判断 position 是否小于 limit，即判断 ByteBuffer 中的字节流是否全部写入到通道中。
            while (this.byteBufferRead.hasRemaining()) {
                try {
                    //这个地方是把主节点发送过来的数据写入到this.byteBufferRead这个缓存中
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    //判断读到的数据长度是否大于0
                    if (readSize > 0) {
                        readSizeZeroTimes = 0;
                        //分发read请求,不断的一批一批的分发处理
                        boolean result = this.dispatchReadRequest();
                        if (!result) {
                            log.error("HAClient, dispatchReadRequest error");
                            return false;
                        }
                    } else if (readSize == 0) {
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        log.info("HAClient, processReadEvent read socket < 0");
                        return false;
                    }
                } catch (IOException e) {
                    log.info("HAClient, processReadEvent read socket exception", e);
                    return false;
                }
            }

            return true;
        }

        private boolean dispatchReadRequest() {
            //主服务器这次传输过来的这次偏移量以及数据量大小
            final int msgHeaderSize = 8 + 4; // phyoffset + size
            //byteBufferRead变量的position值表示从SocketChannel中收到的数据的最后位置；
            //首先将此position值赋值给临时变量readSocketPos，
            int readSocketPos = this.byteBufferRead.position();
            //用HAClient.dispatchPostion变量来标记从byteBufferRead变量读取数据的位置；初始化值为0；
            while (true) {
                //比较position减dispatchPostion的值大于12（消息头部长度为12个字节）：
                //dispatchPosition初始化为0
                int diff = this.byteBufferRead.position() - this.dispatchPosition;
                //若大于12个字节表示有心跳消息从主用Broker发送过来
                /**
                 * 在byteBufferRead中从dispatchPostion位置开始读取数据，初始化状态下dispatchPostion等于0；
                 * 读取8个字节的数据即为主用Broker的同步的起始物理偏移量masterPhyOffset，
                 * 再后4字节为数据的大小bodySize；
                 */
                if (diff >= msgHeaderSize) {
                    //拿到主服务器的最大物理偏移量
                    long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPosition);
                    //拿到body的数据大小
                    int bodySize = this.byteBufferRead.getInt(this.dispatchPosition + 8);

                    //拿到从服务器的最大物理偏移量
                    long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();

                    //不等于0表示之前已经同步过数据到从服务器
                    if (slavePhyOffset != 0) {
                        //为啥要做这个判断？他们不相等怎么就返回了？？没明白·        `
                        if (slavePhyOffset != masterPhyOffset) {
                            log.error("master pushed offset not equal the max phy offset in slave, SLAVE: "
                                + slavePhyOffset + " MASTER: " + masterPhyOffset);
                            return false;
                        }
                    }

                    //说明传递过来的不仅有心跳消息头信息。还是消息本身的信息
                    if (diff >= (msgHeaderSize + bodySize)) {
                        byte[] bodyData = new byte[bodySize];
                        //把position的位置向后移到msgHeaderSize的位置，也就是移12个bit
                        this.byteBufferRead.position(this.dispatchPosition + msgHeaderSize);
                        //读取消息本身的数据到数组里面去
                        this.byteBufferRead.get(bodyData);

                        //将消息本身内容写到commitlog文件中去，完成了主从消息同步
                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);

                        /**
                         * 将byteBufferRead变量的position值重置为readSocketPos；
                         * dispatchPostion值累计12+bodySize；
                         */
                        this.byteBufferRead.position(readSocketPos);
                        this.dispatchPosition += msgHeaderSize + bodySize;

                        /**
                         * 再次上报偏移量给主broker
                         */
                        if (!reportSlaveMaxOffsetPlus()) {
                            return false;
                        }

                        continue;
                    }
                }

                /**
                 * 若小于12个字节，并且byteBufferRead变量中没有可写空间（this.position>=this.limit）,
                 * 则调用HAClient.reallocateByteBuffer()方法进行ByteBuffer的整理，
                 * 整理过程如下：
                 */
                if (!this.byteBufferRead.hasRemaining()) {
                    this.reallocateByteBuffer();
                }

                break;
            }

            return true;
        }

        private boolean reportSlaveMaxOffsetPlus() {
            boolean result = true;
            //调用DefaultMessageStore.getMaxPhyOffset()方法获取备用Broker本地的最大写入位置即最大物理偏移量
            long currentPhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
            //这个地方很大可能已经大于了
            //因为currentReportedOffset这个值是在这个类的680行赋值的
            //而在这个类的606行已经有新的消息写进commitlog日志文件了
            //所以很有可能已经大于了
            if (currentPhyOffset > this.currentReportedOffset) {
                //把新的位置赋值给当前这个位置变量
                this.currentReportedOffset = currentPhyOffset;
                //把最新的位置上报给主节点
                result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                if (!result) {
                    this.closeMaster();
                    log.error("HAClient, reportSlaveMaxOffset error, " + this.currentReportedOffset);
                }
            }

            return result;
        }

        private boolean connectMaster() throws ClosedChannelException {
            if (null == socketChannel) {
                //拿到主节点地址
                String addr = this.masterAddress.get();
                //如果主节点地址不为空
                if (addr != null) {
                    //通过主节点字符串地点拿到socketAddress对象
                    SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
                    //如果socketAddress对象不为空
                    if (socketAddress != null) {
                        //新建一个从服务器的socketChannel
                        this.socketChannel = RemotingUtil.connect(socketAddress);
                        //如果从服务器的socketChannel不为空
                        if (this.socketChannel != null) {
                            //把这个socketChannel注册到selector上。并且设置对读事件感兴趣
                            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
                        }
                    }
                }

                //获取当前这个broker上的commitlog中映射文件的最大物理偏移量
                this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                //更新上次写入时间就等于当前时间
                this.lastWriteTimestamp = System.currentTimeMillis();
            }

            return this.socketChannel != null;
        }

        private void closeMaster() {
            if (null != this.socketChannel) {
                try {

                    SelectionKey sk = this.socketChannel.keyFor(this.selector);
                    if (sk != null) {
                        sk.cancel();
                    }

                    this.socketChannel.close();

                    this.socketChannel = null;
                } catch (IOException e) {
                    log.warn("closeMaster exception. ", e);
                }

                //上次上报时间清0
                this.lastWriteTimestamp = 0;
                //分发位置清0
                this.dispatchPosition = 0;
                //2个缓存都清0
                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);

                this.byteBufferRead.position(0);
                this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);
            }
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {
                try {
                    //从服务器连接主服务器
                    if (this.connectMaster()) {
                        //判断是否到了该上报时间，也就是上次上报时间到当前时间是否超过5秒钟
                        if (this.isTimeToReportOffset()) {
                            //上报MaxOffset到主broker,一开始currentReportedOffset=0;在reportSlaveMaxOffsetPlus()中给它赋值
                            boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                            //如果上报不成功，关闭socketChannel等一系列动作
                            if (!result) {
                                this.closeMaster();
                            }
                        }

                        // 每隔一秒钟轮询一遍
                        this.selector.select(1000);

                        // 处理主服务器发送过来的消息,是WriteSocketService发送的消息同步消息，解析消息并保存到commitlog中
                        boolean ok = this.processReadEvent();
                        //如果处理主节点发送过来的消息不成功，关闭socketChannel等一系列动作
                        if (!ok) {
                            this.closeMaster();
                        }
                        //拿到备broker最大物理偏移量。。上报给主broker,更新最后上报时间
                        if (!reportSlaveMaxOffsetPlus()) {
                            continue;
                        }

                        /**
                         * 拿到上报间隔时间
                         */
                        long interval =
                            HAService.this.getDefaultMessageStore().getSystemClock().now()
                                - this.lastWriteTimestamp;
                        //如果间隔时间大于配置的间隔时间。则表示这个client已经超时
                        //检查上次写入时间戳lastWriteTimestamp距离现在是否已经过了5秒，即
                        if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig()
                            .getHaHousekeepingInterval()) {
                            log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress
                                + "] expired, " + interval);
                            this.closeMaster();
                            log.warn("HAClient, master not response some time, so close connection");
                        }
                    } else {
                        //如果连接主节点失败，暂停这个线程5S再次执行
                        this.waitForRunning(1000 * 5);
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                    this.waitForRunning(1000 * 5);
                }
            }

            log.info(this.getServiceName() + " service end");
        }
        // private void disableWriteFlag() {
        // if (this.socketChannel != null) {
        // SelectionKey sk = this.socketChannel.keyFor(this.selector);
        // if (sk != null) {
        // int ops = sk.interestOps();
        // ops &= ~SelectionKey.OP_WRITE;
        // sk.interestOps(ops);
        // }
        // }
        // }
        // private void enableWriteFlag() {
        // if (this.socketChannel != null) {
        // SelectionKey sk = this.socketChannel.keyFor(this.selector);
        // if (sk != null) {
        // int ops = sk.interestOps();
        // ops |= SelectionKey.OP_WRITE;
        // sk.interestOps(ops);
        // }
        // }
        // }

        @Override
        public String getServiceName() {
            return HAClient.class.getSimpleName();
        }
    }
}
